package com.heima.article.listen;

import com.alibaba.fastjson.JSON;
import com.heima.article.config.stream.KafkaStreamListener;
import com.heima.common.constants.article.HotArticleConstants;
import com.heima.model.mess.app.ArticleVisitStreamMess;
import com.heima.model.mess.app.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;
import org.springframework.stereotype.Component;

/**
 * @作者 itcast
 * @创建日期 2021/2/24 15:16
 **/
@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String,String>> {
    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC;
    }
    @Override
    public String sendTopic() {
        return HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC;
    }
    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        return stream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                log.info(" 流式处理开始工作 key==>{}   value==>{}",key,value);
                // {type: VIEWS, articleId:123,add:1}
                UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);
                // 文章ID:行为    key: 12432432342:LIKES
                return new KeyValue<>(updateArticleMess.getArticleId()+":"+updateArticleMess.getType(),key);
            }
        }).groupByKey()
          .windowedBy(TimeWindows.of(10000))
          .count() // 12432432342:LIKES 1000      12432432342:COLLECTION 100
          .toStream()
          .map(new KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>() {
              @Override
              public KeyValue<String, String> apply(Windowed<String> stringWindowed, Long count) {
                  String key = stringWindowed.key(); // 12432432342:COLLECTION  100
                  String[] split = key.split(":");
                  // 将统计好的key进行切割
                  Long articleId = Long.valueOf(split[0]);
                  String behavior = split[1];
                  // 封装成消息对象
                  ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
                  mess.setArticleId(articleId);
                  // 新增点赞数量
                  if (behavior.equals(UpdateArticleMess.UpdateArticleType.LIKES.name())) {
                      mess.setLike(count);
                  }
                  // 新增收藏数量
                  if (behavior.equals(UpdateArticleMess.UpdateArticleType.COLLECTION.name())) {
                      mess.setCollect(count);
                  }
                  // 新增阅读数量
                  if (behavior.equals(UpdateArticleMess.UpdateArticleType.VIEWS.name())) {
                      mess.setView(count);
                  }
                  // 新增评论数量
                  if (behavior.equals(UpdateArticleMess.UpdateArticleType.COMMENT.name())) {
                      mess.setComment(count);
                  }
                  log.info(" 流式处理完毕   新的消息  value==>{}",JSON.toJSONString(mess));
                  return new KeyValue<>("",JSON.toJSONString(mess));
              }
          });

    }
}
